package com.study.iceberg.flink;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.data.RowData;
import org.apache.iceberg.flink.TableLoader;
import org.apache.iceberg.flink.source.IcebergSource;
import org.apache.iceberg.flink.source.StreamingStartingStrategy;
import org.apache.iceberg.flink.source.assigner.SimpleSplitAssignerFactory;

/**
 * FLIP27方式读取
 * FLIP-27 (Flink Improvement Proposal 27) 是 Apache Flink 社区中的一个改进提案，旨在提高 Flink 对批处理任务的支持和性能。该提案的主要目标是将 Flink 的批处理引擎与其流式计算引擎进行整合，以实现更好的资源利用和更高的批处理性能。
 * <p>
 * 具体来说，FLIP-27 提出了以下几个改进点：
 * <p>
 * 批处理引擎和流式计算引擎的整合，使得批处理任务能够使用流式计算引擎中的优化器、运行时和资源管理器等组件。
 * <p>
 * 引入“数据倾斜度自适应调整”功能，以自动检测和处理批处理作业中的数据倾斜问题。
 * <p>
 * 改进任务调度和资源分配机制，以更好地支持批处理场景。
 * <p>
 * 改进状态后端和存储机制，以提高批处理作业的吞吐量和延迟表现。
 * <p>
 * 引入新的 Table API 和 SQL 接口，以更好地支持批处理场景下的数据处理。
 * <p>
 * FLIP-27 是 Flink 社区中一个重要的改进提案，它将进一步提升 Flink 在批处理场景中的性能和可用性，使得 Flink 能够更好地满足企业级数据处理需求。
 */
public class ReadDemo2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://192.168.200.101:9820/warehouse/iceberg-hadoop/iceberg_db/sample");

        // 获取batch模式的数据源
        /*IcebergSource<RowData> source = IcebergSource.forRowData()
                .tableLoader(tableLoader)
                .assignerFactory(new SimpleSplitAssignerFactory())
                .build();*/

        // 获取streaming模式的数据源
        IcebergSource<RowData> source = IcebergSource.forRowData()
                .tableLoader(tableLoader)
                .assignerFactory(new SimpleSplitAssignerFactory())
                .streaming(true)
                //指定流式读取策略，这里选从最早的数据开始读取
                .streamingStartingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_EARLIEST_SNAPSHOT)
                .build();


        DataStreamSource<RowData> inputDS = env.fromSource(
                source,
                WatermarkStrategy.noWatermarks(),
                "iceberg source",
                TypeInformation.of(RowData.class)
        );

        inputDS.map(r -> Tuple2.of(r.getInt(0), r.getString(1).toString()))
                .returns(Types.TUPLE(Types.INT, Types.STRING))
                .print();

        env.execute();
    }
}
